package com.xl.bigdata.tools.kafka.producer;

import com.xl.bigdata.tools.date.DateTools;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @ClassName KafkaProducerTool
 * @Description TODO
 * @Author xl.gao
 * @Date 2022/2/8 14:50
 * @Version
 */
public class KafkaProducer86Tool {

    private static volatile KafkaProducer<String, String> producer;

    private static Properties props;

    /**
     * 获得链接
     */
    public static synchronized KafkaProducer<String, String> getKafkaProducer(String brokers, String threadName) {

        try {
            if (producer == null) {
                synchronized (KafkaProducer86Tool.class){
                    if(producer == null){
                        props = new Properties();
                        //props.put("bootstrap.servers", ""lx-es-04:9092,lx-es-05:9092"");
                        props.put("bootstrap.servers", brokers);
                        props.put("acks", "all");
                        props.put("retries", 0);
                        props.put("batch.size", 16384);
                        props.put("key.serializer", StringSerializer.class.getName());
                        props.put("value.serializer", StringSerializer.class.getName());
                        producer = new KafkaProducer<String, String>(props);
                    }
                }

                System.out.println("running-"+ DateTools.getNowDate() +"-"+threadName+"-KafkaProducerClient - 获取KafkaProducerClient连接成功...");
            } else {
                System.out.println("running-"+ DateTools.getNowDate() +"-"+threadName+"-KafkaProducerClient - 重复使用KafkaProducerClient连接...");
            }
        } catch (Exception e) {
            System.out.println("running-"+ DateTools.getNowDate() +"-"+threadName+"-KafkaProducerClient - 获取kafka连接失败...");
            e.printStackTrace();
        }

        return producer;

    }

    public synchronized void close(KafkaProducer producer) {
        if (producer != null) {
            producer.close();
        }
    }

    public static synchronized void sendData(KafkaProducer producer,String url, String topic, String key, String value) {
        try {
            if (producer != null) {
                producer.send(new ProducerRecord<String, String>(topic, key, value));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
